#include "../include/common.h"
//#include <gflags/gflags.h>
#include <gtest/gtest.h>
#include <chrono>
#define N 5000000
#define MASTER "192.168.50.10"

std::chrono::system_clock::time_point global_start;
std::chrono::system_clock::time_point global_end;


void wait_replies_for_throughput_test(rdma_fd *handler, int n){
//    ProposeReplyTS reply;
    serializeBuff *message = new serializeBuff();
    for(int i = 0; i < n; i++){
        uint8_t msg_type = 0;
        message->len = 0;
        message->buff = NULL;
        if(handler->have_read + sizeof(uint8_t) + sizeof(uint32_t) + 1 > handler->receive_buf_size){
            handler->have_read = 0;
        }
        while(1){
            volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + 5);
            if(*end == '1'){
                break;
            }
        }
        memcpy(&msg_type, handler->receive_buf + handler->have_read, sizeof(uint8_t));
        memcpy(&message->len, handler->receive_buf + handler-> have_read + sizeof(uint8_t), sizeof(uint32_t));
        //msg_type = *((uint8_t *)handler->receive_buf + handler->have_read);
        //message->len = *((uint32_t *)handler->receive_buf + handler->have_read + sizeof(uint8_t));
        //printf("1.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
        memset(handler->receive_buf + handler->have_read, 0, sizeof(uint8_t) + sizeof(uint32_t) + 1);
        handler->have_read += sizeof(uint8_t) + sizeof(uint32_t) + 1;
        if(handler->have_read + message->len + 1 > handler->receive_buf_size){
            handler->have_read = 0;
        }
        while(1){
            volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + message->len);
            if(*end == '1'){
                break;  
            }
        }
        //printf("2.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
        message->buff = malloc(message->len);
        memcpy(message->buff, handler->receive_buf + handler->have_read, message->len);
        memset(handler->receive_buf + handler->have_read, 0, message->len + 1);
        handler->have_read += message->len + 1;        
    }
    delete message;
}



TEST(CoordinatorTest,Latency_test){

    rdma_fd *handler = (rdma_fd *)malloc(sizeof(rdma_fd));
    int ret = build_connection(handler, MASTER, MASTER_PORT);
    ASSERT_EQ(ret, 0);
//    printf(argv[1]);
    std::vector<serializeBuff *> data(N, NULL);
    prepare_latency_data_no_conflict(data, N, 10);
//  time begin;
    std::thread *th = new std::thread(wait_replies_for_throughput_test, handler, N);
    sleep(1);
    global_start = std::chrono::system_clock::now();
    for(int i = 0; i < N; i++){
        client_send_message(handler, data[i], PROPOSE);
    }
    th->join();
    global_end = std::chrono::system_clock::now();
    std::chrono::duration<double> diff = global_end-global_start;
    printf("[TIME]: %lf s [Throughput]: %lf op/s\n", diff.count(), (double) N / diff.count() ); 
}





int main(int argc, char * argv[]) {
    testing::InitGoogleTest(&argc, argv);  
    return RUN_ALL_TESTS();  
}